{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Parallel System Tools in [Programming Python, 4th Edition](http://www.rmi.net/~lutz/about-pp4e.html), include forks,\n",
    "threads, pipes, signals, sockets, and other launching techniques.\n",
    "\n",
    "<!-- TEASER_END -->"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Forking Processes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1.fork"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def child():\n",
    "    print \"hello from child\", os.getpid()\n",
    "    os._exit(0) # else goes back to parent loop"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def parent():\n",
    "    while True:\n",
    "        #generates a copy of the calling program, \n",
    "        # it returns a different value in each copy: \n",
    "        # zero in the child process and \n",
    "        # the process ID of the new child in the parent.\n",
    "        newpid = os.fork() \n",
    "        if newpid == 0:\n",
    "            child()\n",
    "        else:\n",
    "            print \"hello from parent\", os.getpid(), newpid\n",
    "        if raw_input() == 'q': break"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hello from parent 8118 8360\n",
      "\n",
      "hello from parent 8118 8366\n",
      "hello from child 8360\n",
      "\n",
      "hello from parent 8118 8373\n",
      "hello from child 8366\n",
      "q\n",
      "hello from child 8373\n"
     ]
    }
   ],
   "source": [
    "parent()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2.fork-count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import os,time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def counter(count):\n",
    "    for i in range(count):\n",
    "        time.sleep(3)\n",
    "        print \"[%s] => %s\" % (os.getpid(), i)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Process 9008 spawned\n",
      "Process 9009 spawned\n",
      "Process 9010 spawned\n",
      "Process 9011 spawned\n",
      "Process 9012 spawned\n",
      "Main process exiting.\n",
      "[8979] => 0\n",
      "[8978] => 0\n",
      "[8980] => 0\n",
      "[8981] => 0\n",
      "[8982] => 0\n",
      "[8979] => 1[8978] => 1[8980] => 1[8981] => 1[8982] => 1\n",
      "\n",
      "\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "for i in range(5):\n",
    "    pid = os.fork()\n",
    "    if pid != 0:\n",
    "        print \"Process %d spawned\" % pid\n",
    "    else:\n",
    "        counter(5)\n",
    "        os._exit(0)\n",
    "print \"Main process exiting.\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3.The fork/exec Combination"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Child is 9654\n",
      "\n",
      "Child is 9655\n",
      "\n",
      "Child is 9656\n",
      "\n",
      "Child is 9657\n",
      "q\n"
     ]
    }
   ],
   "source": [
    "parm = 0\n",
    "while True:\n",
    "    parm += 1\n",
    "    pid = os.fork()\n",
    "    if pid == 0:\n",
    "        os.execlp('python','python','child.py',str(parm))\n",
    "        assert False, 'error starting program'\n",
    "    else:\n",
    "        print \"Child is\", pid\n",
    "        if raw_input() == 'q': break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Threads"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1.The thread Module"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.1Basic usage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import thread"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def child(tid):\n",
    "    print 'Hello from thread', tid"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def parent():\n",
    "    i = 0\n",
    "    while True:\n",
    "        i += 1\n",
    "        thread.start_new_thread(child, (i,))\n",
    "        if raw_input() == 'q': break"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Hello from thread 1\n",
      "\n",
      "Hello from thread 2\n",
      "q\n",
      "Hello from thread 3\n"
     ]
    }
   ],
   "source": [
    "parent()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.2 Other ways to code threads with thread"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import thread"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def action(i):\n",
    "    print i**32"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "class Power:\n",
    "    def __init__(self,i):\n",
    "        self.i = i\n",
    "    def action(self):\n",
    "        print self.i**32"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "140621501368064"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "thread.start_new_thread(action,(2,))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4294967296\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "140621501368064"
      ]
     },
     "execution_count": 52,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "thread.start_new_thread(lambda: action(2), ())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "obj = Power(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "140621501368064"
      ]
     },
     "execution_count": 57,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "thread.start_new_thread(obj.action,())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.3 Running multiple threads"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import thread,time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def counter(myId, count):\n",
    "    for i in range(count):\n",
    "        time.sleep(0.1)\n",
    "        print '[%s] => %s' % (myId, i)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "for i in range(5):\n",
    "    thread.start_new_thread(counter,(i,3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2] => 0\n",
      "[4] => 0\n",
      "[1] => 0\n",
      "[0] => 0\n",
      "[3] => 0\n",
      "[2] => 1\n",
      "[4] => 1\n",
      "[1] => 1\n",
      "[0] => 1\n",
      "[3] => 1\n",
      "[2] => 2\n",
      "[4] => 2\n",
      "[1] => 2\n",
      "[0] => 2\n",
      "[3] => 2\n",
      "Main threading exit.\n"
     ]
    }
   ],
   "source": [
    "time.sleep(2)\n",
    "print 'Main threading exit.'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.4 Synchronizing access to shared objects and names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import thread,time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def counter(myId, count):\n",
    "    for i in range(count):\n",
    "        time.sleep(0.1)\n",
    "        mutex.acquire()\n",
    "        print '[%s] => %s' % (myId, i)\n",
    "        mutex.release()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "mutex = thread.allocate_lock()\n",
    "for i in range(5):\n",
    "    thread.start_new_thread(counter,(i,3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0] => 0\n",
      "[1] => 0\n",
      "[2] => 0\n",
      "[3] => 0\n",
      "[4] => 0\n",
      "[0] => 1\n",
      "[1] => 1\n",
      "[2] => 1\n",
      "[3] => 1\n",
      "[4] => 1\n",
      "[0] => 2\n",
      "[1] => 2\n",
      "[2] => 2\n",
      "[3] => 2\n",
      "[4] => 2\n",
      "Main threading exit.\n"
     ]
    }
   ],
   "source": [
    "time.sleep(2)\n",
    "print 'Main threading exit.'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.5 Waiting for spawned thread exits"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# uses mutexes to know when threads are done in parent/main thread,\n",
    "# instead of time.sleep; lock stdout to avoid comingled prints;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import thread"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "stdoutmutex = thread.allocate_lock()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# exitmutex = [thread.allocate_lock() for i in range(5)]\n",
    "exitmutex = [False] * 5"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def counter(myId, count):\n",
    "    for i in range(count):\n",
    "        time.sleep(0.1)\n",
    "        stdoutmutex.acquire()\n",
    "        print '[%s] => %s' % (myId, i)\n",
    "        stdoutmutex.release()\n",
    "#     exitmutex[myId].acquire() # signal main thread\n",
    "    exitmutex[myId] = True"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "for i in range(5):\n",
    "    thread.start_new_thread(counter,(i,3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0] => 0\n",
      "[1] => 0\n",
      "[2] => 0\n",
      "[3] => 0\n",
      "[4] => 0\n",
      "[0] => 1\n",
      "[1] => 1\n",
      "[2] => 1\n",
      "[3] => 1\n",
      "[4] => 1\n",
      "[0] => 2\n",
      "[1] => 2\n",
      "[2] => 2\n",
      "[4] => 2\n",
      "[3] => 2\n",
      "Main threading exit.\n"
     ]
    }
   ],
   "source": [
    "for mutex in exitmutex:\n",
    "#     while not mutex.locked():pass\n",
    "    while not True:pass\n",
    "print 'Main threading exit.'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2.The threading Module"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.1 thread-classes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import threading"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "class mythread(threading.Thread):          # subclass Thread object\n",
    "    def __init__(self, myId, count):\n",
    "        self.myId = myId\n",
    "        self.count = count\n",
    "        threading.Thread.__init__(self)\n",
    "    def run(self):\n",
    "        for i in range(self.count):\n",
    "            stdoutmutex.acquire()\n",
    "            print '[%s] => %s' % (self.myId, i)\n",
    "            stdoutmutex.release()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "stdoutmutex = threading.Lock() # same as thread.allocate_lock( )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0] => 0\n",
      "[0] => 1\n",
      "[0] => 2\n",
      "[1] => 0\n",
      "[1] => 1\n",
      "[1] => 2\n",
      "[2] => 0\n",
      "[2] => 1\n",
      "[2] => 2\n",
      "[3] => 0\n",
      "[3] => 1\n",
      "[3] => 2\n",
      "[4] => 0\n",
      "[4] => 1\n",
      "[4] => 2\n",
      "Main threading exit.\n"
     ]
    }
   ],
   "source": [
    "threads = []\n",
    "for i in range(5):\n",
    "    thread = mythread(i, 3)\n",
    "    thread.start()\n",
    "    threads.append(thread)\n",
    "for thread in threads:\n",
    "    thread.join()\n",
    "print 'Main threading exit.'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.2 Other ways to code threads with threading"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4294967296\n",
      "4294967296\n"
     ]
    }
   ],
   "source": [
    "import threading\n",
    "# subclass with state\n",
    "class mythread(threading.Thread):\n",
    "    def __init__(self, i):\n",
    "        self.i = i\n",
    "        threading.Thread.__init__(self)\n",
    "    def run(self):\n",
    "        print self.i**32\n",
    "mythread(2).start( )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def action(i):\n",
    "    print i**32"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# pass action in\n",
    "thread = threading.Thread(target=(lambda: action(i)))\n",
    "thread.start( )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "140349501798144"
      ]
     },
     "execution_count": 98,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# same but no lambda wrapper for state\n",
    "threading.Thread(target=action, args=(2,)).start( )\n",
    "\n",
    "# basic thread module\n",
    "import thread\n",
    "thread.start_new_thread(action, (2,))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4294967296\n",
      "4294967296\n"
     ]
    }
   ],
   "source": [
    "# a non-thread class with state, OOP\n",
    "class Power:\n",
    "    def __init__(self, i):\n",
    "        self.i = i\n",
    "    def action(self):\n",
    "        print(self.i ** 32)\n",
    "obj = Power(2)\n",
    "threading.Thread(target=obj.action).start() # thread runs bound method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "140349510190848"
      ]
     },
     "execution_count": 101,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# nested scope to retain state\n",
    "def action(i):\n",
    "    def power():\n",
    "        print(i ** 32)\n",
    "    return power\n",
    "threading.Thread(target=action(2)).start() # thread runs returned function\n",
    "# both with basic thread module\n",
    "thread.start_new_thread(obj.action, ())\n",
    "thread.start_new_thread(action(2), ()) # thread runs a callable object"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.3 Synchronizing access to shared objects and names revisited"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "200\n"
     ]
    }
   ],
   "source": [
    "import threading\n",
    "count = 0\n",
    "def adder():\n",
    "    global count\n",
    "    count += 1\n",
    "#     time.sleep(0.5)\n",
    "    count += 1\n",
    "threads = []\n",
    "for i in range(100):\n",
    "    thread = threading.Thread(target=adder, args=())\n",
    "    thread.start()\n",
    "    threads.append(thread)\n",
    "for thread in threads: thread.join()\n",
    "print count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 130,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "200\n"
     ]
    }
   ],
   "source": [
    "#prints 200 each time, because shared resource access synchronized\n",
    "count = 0\n",
    "def adder(addlock):\n",
    "    global count\n",
    "    with addlock:\n",
    "        count += 1\n",
    "#     time.sleep(0.5)\n",
    "    with addlock:\n",
    "        count += 1\n",
    "addlock = threading.Lock()\n",
    "threads = []\n",
    "for i in range(100):\n",
    "    thread = threading.Thread(target=adder, args=(addlock,))\n",
    "    thread.start()\n",
    "    threads.append(thread)\n",
    "for thread in threads: thread.join()\n",
    "print count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# The queue Module"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#producer and consumer threads communicating with a shared queue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "numconsumers = 2 # how many consumers to start\n",
    "numproducers = 4  # how many producers to start\n",
    "nummessages = 4  # messages per producer to put"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "consumer 0 got => [producer id=0, count=0]\n",
      "consumer 1 got => [producer id=0, count=1]\n",
      "consumer 0 got => [producer id=0, count=2]\n",
      "consumer 1 got => [producer id=0, count=3]\n",
      "consumer 0 got => [producer id=1, count=0]\n",
      "consumer 0 got => [producer id=2, count=0]\n",
      "consumer 1 got => [producer id=1, count=1]\n",
      "consumer 0 got => [producer id=3, count=0]\n",
      "consumer 1 got => [producer id=1, count=2]\n",
      "consumer 0 got => [producer id=2, count=1]\n",
      "consumer 1 got => [producer id=1, count=3]\n",
      "consumer 0 got => [producer id=2, count=2]\n",
      "consumer 1 got => [producer id=3, count=1]\n",
      "consumer 0 got => [producer id=2, count=3]\n",
      "consumer 0 got => [producer id=3, count=2]\n",
      "consumer 0 got => [producer id=3, count=3]\n",
      "Main thread exit.\n"
     ]
    }
   ],
   "source": [
    "import thread, Queue, time\n",
    "safeprint = thread.allocate_lock()  # else print may overlap\n",
    "dataQueue = Queue.Queue()        # shared global. infinite size\n",
    "\n",
    "def producer(idnum):\n",
    "    for msgnum in range(nummessages):\n",
    "        time.sleep(idnum)\n",
    "        dataQueue.put('[producer id=%d, count=%d]' % (idnum, msgnum))\n",
    "def consumer(idnum):\n",
    "    while True:\n",
    "        time.sleep(0.1)\n",
    "        try:\n",
    "            data = dataQueue.get(block=False)\n",
    "        except Queue.Empty:\n",
    "            pass\n",
    "        else:           \n",
    "            safeprint.acquire( )\n",
    "            print 'consumer', idnum, 'got =>', data\n",
    "            safeprint.release( )\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    for i in range(numconsumers):\n",
    "        thread.start_new_thread(consumer, (i,))\n",
    "    for i in range(numproducers):\n",
    "        thread.start_new_thread(producer, (i,))\n",
    "    time.sleep(((numproducers-1) * nummessages) + 1)\n",
    "    print('Main thread exit.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "consumer 1 got => [producer id=0, count=0]\n",
      "consumer 0 got => [producer id=0, count=1]\n",
      "consumer 1 got => [producer id=0, count=2]\n",
      "consumer 0 got => [producer id=0, count=3]\n",
      "consumer 0 got => [producer id=1, count=0]\n",
      "consumer 1 got => [producer id=2, count=0]\n",
      "consumer 0 got => [producer id=1, count=1]\n",
      "consumer 1 got => [producer id=3, count=0]\n",
      "consumer 0 got => [producer id=1, count=2]\n",
      "consumer 1 got => [producer id=2, count=1]\n",
      "consumer 1 got => [producer id=1, count=3]\n",
      "consumer 1 got => [producer id=3, count=1]\n",
      "consumer 1 got => [producer id=2, count=2]\n",
      "consumer 1 got => [producer id=2, count=3]\n",
      "consumer 1 got => [producer id=3, count=2]\n",
      "consumer 1 got => [producer id=3, count=3]\n",
      "Main thread exit.\n"
     ]
    }
   ],
   "source": [
    "#same as queuetest.py, by queue object pass in as argument, not global\n",
    "numconsumers = 2                  # how many consumers to start\n",
    "numproducers = 4                  # how many producers to start\n",
    "nummessages  = 4                  # messages per producer to put\n",
    "\n",
    "import thread, Queue, time\n",
    "safeprint = thread.allocate_lock()    # else prints may overlap\n",
    "dataQueue = Queue.Queue()             # shared global, infinite size\n",
    "\n",
    "def producer(idnum, dataqueue):\n",
    "    for msgnum in range(nummessages):\n",
    "        time.sleep(idnum)\n",
    "        dataqueue.put('[producer id=%d, count=%d]' % (idnum, msgnum))\n",
    "\n",
    "def consumer(idnum, dataqueue):\n",
    "    while True:\n",
    "        time.sleep(0.1)\n",
    "        try:\n",
    "            data = dataqueue.get(block=False)\n",
    "        except Queue.Empty:\n",
    "            pass\n",
    "        else:\n",
    "            with safeprint:\n",
    "                print 'consumer', idnum, 'got =>', data\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    for i in range(numconsumers):\n",
    "        thread.start_new_thread(consumer, (i, dataQueue))\n",
    "    for i in range(numproducers):\n",
    "        thread.start_new_thread(producer, (i, dataQueue))\n",
    "    time.sleep(((numproducers-1) * nummessages) + 1)\n",
    "    print 'Main thread exit.'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "consumer 0 got => [producer id=3, count=3]\n",
      "consumer 1 got => [producer id=0, count=0]\n",
      "consumer 0 got => [producer id=0, count=1]\n",
      "consumer 0 got => [producer id=0, count=2]\n",
      "consumer 1 got => [producer id=0, count=3]\n",
      "consumer 1 got => [producer id=1, count=0]\n",
      "consumer 0 got => [producer id=1, count=1]\n",
      "consumer 1 got => [producer id=2, count=0]\n",
      "consumer 0 got => [producer id=1, count=2]\n",
      "consumer 1 got => [producer id=3, count=0]\n",
      "consumer 0 got => [producer id=1, count=3]\n",
      "consumer 0 got => [producer id=2, count=1]\n",
      "consumer 0 got => [producer id=2, count=2]\n",
      "consumer 0 got => [producer id=3, count=1]\n",
      "consumer 0 got => [producer id=2, count=3]\n",
      "consumer 0 got => [producer id=3, count=2]\n",
      "Main thread exit.\n"
     ]
    }
   ],
   "source": [
    "#same as queuetest2.py, but uses threading, not _threads\n",
    "numconsumers = 2                  # how many consumers to start\n",
    "numproducers = 4                  # how many producers to start\n",
    "nummessages  = 4                  # messages per producer to put\n",
    "\n",
    "import threading, Queue, time\n",
    "safeprint = threading.Lock()          # else prints may overlap\n",
    "dataQueue = Queue.Queue()             # shared global, infinite size\n",
    "\n",
    "def producer(idnum, dataqueue):\n",
    "    for msgnum in range(nummessages):\n",
    "        time.sleep(idnum)\n",
    "        dataqueue.put('[producer id=%d, count=%d]' % (idnum, msgnum))\n",
    "\n",
    "def consumer(idnum, dataqueue):\n",
    "    while True:\n",
    "        time.sleep(0.1)\n",
    "        try:\n",
    "            data = dataqueue.get(block=False)\n",
    "        except Queue.Empty:\n",
    "            pass\n",
    "        else:\n",
    "            with safeprint:\n",
    "                print 'consumer', idnum, 'got =>', data\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    for i in range(numconsumers):\n",
    "        thread = threading.Thread(target=consumer, args=(i, dataQueue))\n",
    "        thread.daemon = True  # else cannot exit!\n",
    "        thread.start()\n",
    "\n",
    "    waitfor = []\n",
    "    for i in range(numproducers):\n",
    "        thread = threading.Thread(target=producer, args=(i, dataQueue))\n",
    "        waitfor.append(thread)\n",
    "        thread.start()\n",
    "\n",
    "    for thread in waitfor: thread.join()    # or time.sleep() long enough here\n",
    "    print 'Main thread exit.'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Program Exits"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1.Process Exit Status and Shared State"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "exitstat = 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def child():\n",
    "    global exitstat\n",
    "    exitstat += 1\n",
    "    print 'Hello from child', os.getpid(), exitstat\n",
    "    os._exit(exitstat)\n",
    "    print 'never reached'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def parent():\n",
    "    while True:\n",
    "        newpid = os.fork()\n",
    "        if newpid == 0:\n",
    "            child()\n",
    "        else:\n",
    "            pid, status = os.wait()\n",
    "            print 'Parent got', pid, status, (status >> 8)\n",
    "            if raw_input() == 'q': break"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Parent got 4367 256 1\n",
      "Hello from child 4367 1\n",
      "\n",
      "Parent got 4373 256 1\n",
      "Hello from child 4373 1\n",
      "\n",
      "Parent got 4379 256 1\n",
      "Hello from child 4379 1\n",
      "\n",
      "Parent got 4385 256 1\n",
      "Hello from child 4385 1\n",
      "\n",
      "Parent got 4391 256 1\n",
      "Hello from child 4391 1\n",
      "q\n"
     ]
    }
   ],
   "source": [
    "parent()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Thread Exits and Shared State"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import thread"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "exitstat = 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Hello from child 140613521495808 1\n",
      "\n",
      "Hello from child 140613521495808 2\n",
      "q\n",
      "Hello from child 140613521495808 3\n"
     ]
    }
   ],
   "source": [
    "def child():\n",
    "    global exitstat\n",
    "    exitstat += 1\n",
    "    threadid = thread.get_ident()\n",
    "    print 'Hello from child', threadid, exitstat\n",
    "    thread.exit()\n",
    "    print 'never reached'\n",
    "def parent():\n",
    "    while True:\n",
    "        thread.start_new_thread(child, ())\n",
    "        if raw_input() == 'q': break\n",
    "if __name__ == '__main__': parent()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# As we’ve learned, a thread normally exits silently when the function it runs returns,\n",
    "# and the function return value is ignored. Optionally, the _thread.exit function can be\n",
    "# called to terminate the calling thread explicitly and silently. This call works almost\n",
    "# exactly like sys.exit (but takes no return status argument), and it works by raising a\n",
    "# SystemExit exception in the calling thread. Because of that, a thread can also prema-\n",
    "# turely end by calling sys.exit or by directly raising SystemExit . Be sure not to call\n",
    "# os._exit within a thread function, though—doing so can have odd results (the last time\n",
    "# I tried, it hung the entire process on my Linux system and killed every thread in the\n",
    "# process on Windows!)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# keep in mind that threads and processes have default lifespan models,\n",
    "# which we explored earlier. By way of review, when child threads are still running, the\n",
    "# two thread modules’ behavior differs—programs on most platforms exit when the pa-\n",
    "# rent thread does under _thread , but not normally under threading unless children are\n",
    "# made daemons. When using processes, children normally outlive their parent. This\n",
    "# different process behavior makes sense if you remember that threads are in-process\n",
    "# function calls, but processes are more independent and autonomous.\n",
    "# When used well, exit status can be used to implement error detection and simple com-\n",
    "# munication protocols in systems composed of command-line scripts. But having said\n",
    "# that, I should underscore that most scripts do simply fall off the end of the source to\n",
    "# exit, and most thread functions simply return; explicit exit calls are generally employed\n",
    "# for exceptional conditions and in limited contexts only."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Interprocess Communication"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Anonymous pipe"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.1 Anonymous pipe basics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import os, time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def child(pipeout):\n",
    "    zzz = 0\n",
    "    while True:\n",
    "        time.sleep(zzz)                                    #make parent wait\n",
    "        msg = ('Spam %03d' % zzz).encode() #pipes are binary bytes\n",
    "        os.write(pipeout, msg)                       #send to parent\n",
    "        zzz = (zzz+1) % 5                                #goto 0 after 4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def parent():\n",
    "    pipein,pipeout = os.pipe()\n",
    "    if os.fork == 0:\n",
    "        child(pipeout)\n",
    "    else:\n",
    "        while True:\n",
    "            line = os.read(pipein, 32)\n",
    "            # blocks until data sent\n",
    "            print 'Parent %d got [%s] at %s' % (os.getpid(), line, time.time())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "parent()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.2 Wrapping pipe descriptors in file objects"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# same as pipe1.py, but wrap pipe input in stdio file object\n",
    "# to read by line, and close unused pipe fds in both processes\n",
    "import os, time\n",
    "def child(pipeout):\n",
    "    zzz = 0\n",
    "    while True:\n",
    "        time.sleep(zzz)\n",
    "        msg = ('Spam %03d\\n' % zzz).encode()\n",
    "        os.write(pipeout, msg)\n",
    "        zzz = (zzz+1) % 5\n",
    "def parent():\n",
    "    pipein, pipeout = os.pipe()\n",
    "    if os.fork() == 0:\n",
    "        os.close(pipein)\n",
    "        child(pipeout)\n",
    "    else:\n",
    "        os.close(pipeout)\n",
    "        pipein = os.fdopen(pipein)\n",
    "        while True:\n",
    "            line = pipein.readline()[:-1]\n",
    "            print('Parent %d got [%s] at %s' % (os.getpid(), line, time.time()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "parent()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.1.3 Anonymous pipes and threads"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import os, time, threading"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def child(pipeout):\n",
    "    zzz = 0\n",
    "    while True:\n",
    "        time.sleep(zzz)\n",
    "        msg = (\"Spam %03d\" % zzz).encode()\n",
    "        os.write(pipeout, msg)\n",
    "        zzz = (zzz +1) % 5"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def parent(pipein):\n",
    "    while True:\n",
    "        line = os.read(pipein, 32)\n",
    "        print 'Parent %d got [%s] at %s' % (os.getpid(), line, time.time())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "pipein ,pipeout = os.pipe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "threading.Thread(target=child, args=(pipeout,)).start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Parent 3257 got [Spam 003Spam 004Spam 000Spam 001] at 1400549465.95\n",
      "Parent 3257 got [Spam 002Spam 003Spam 004Spam 000] at 1400549465.95\n",
      "Parent 3257 got [Spam 001Spam 002Spam 003] at 1400549465.95\n",
      "Parent 3257 got [Spam 004] at 1400549468.75\n",
      "Parent 3257 got [Spam 000] at 1400549468.76\n",
      "Parent 3257 got [Spam 001] at 1400549469.76\n",
      "Parent 3257 got [Spam 002] at 1400549471.76\n",
      "Parent 3257 got [Spam 003] at 1400549474.76\n"
     ]
    },
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m\n\u001b[1;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-7-a47c722744b8>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[0mparent\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpipein\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;32m<ipython-input-3-218f40ee8ea0>\u001b[0m in \u001b[0;36mparent\u001b[1;34m(pipein)\u001b[0m\n\u001b[0;32m      1\u001b[0m \u001b[1;32mdef\u001b[0m \u001b[0mparent\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpipein\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      2\u001b[0m     \u001b[1;32mwhile\u001b[0m \u001b[0mTrue\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 3\u001b[1;33m         \u001b[0mline\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mos\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mread\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpipein\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m32\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      4\u001b[0m         \u001b[1;32mprint\u001b[0m \u001b[1;34m'Parent %d got [%s] at %s'\u001b[0m \u001b[1;33m%\u001b[0m \u001b[1;33m(\u001b[0m\u001b[0mos\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgetpid\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mline\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mtime\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mtime\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "parent(pipein)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.1.4 Bidirectional IPC with anonymous pipes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# spawn a child process/program, connect my stdin/stdout to child process's\n",
    "# stdout/stdin--my reads and writes map to output and input streams of the\n",
    "# spawned program; much like tying together streams with subprocess module;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import sys, os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "ename": "UnsupportedOperation",
     "evalue": "IOStream has no fileno.",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m\n\u001b[1;31mUnsupportedOperation\u001b[0m                      Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-19-8b7123926819>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m     22\u001b[0m \u001b[1;32mif\u001b[0m \u001b[0m__name__\u001b[0m \u001b[1;33m==\u001b[0m \u001b[1;34m'__main__'\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     23\u001b[0m     \u001b[0mmypid\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mos\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgetpid\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 24\u001b[1;33m     \u001b[0mspawn\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'python'\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;34m'-u'\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;34m'pipes-testchild.py'\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;34m'spam'\u001b[0m\u001b[1;33m)\u001b[0m     \u001b[1;31m# fork child program\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     25\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     26\u001b[0m     \u001b[1;32mprint\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'Hello 1 from parent'\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mmypid\u001b[0m\u001b[1;33m)\u001b[0m               \u001b[1;31m# to child's stdin\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m<ipython-input-19-8b7123926819>\u001b[0m in \u001b[0;36mspawn\u001b[1;34m(prog, *args)\u001b[0m\n\u001b[0;32m      1\u001b[0m \u001b[1;32mdef\u001b[0m \u001b[0mspawn\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mprog\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m*\u001b[0m\u001b[0margs\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m                       \u001b[1;31m# pass progname, cmdline args\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      2\u001b[0m     \u001b[0mstdinFd\u001b[0m  \u001b[1;33m=\u001b[0m \u001b[0msys\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstdin\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mfileno\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m             \u001b[1;31m# get descriptors for streams\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 3\u001b[1;33m     \u001b[0mstdoutFd\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0msys\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstdout\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mfileno\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m            \u001b[1;31m# normally stdin=0, stdout=1\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      4\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      5\u001b[0m     \u001b[0mparentStdin\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mchildStdout\u001b[0m  \u001b[1;33m=\u001b[0m \u001b[0mos\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mpipe\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m     \u001b[1;31m# make two IPC pipe channels\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/usr/local/lib/python2.7/dist-packages/IPython/kernel/zmq/iostream.pyc\u001b[0m in \u001b[0;36mfileno\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    185\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    186\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mfileno\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 187\u001b[1;33m         \u001b[1;32mraise\u001b[0m \u001b[0mUnsupportedOperation\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"IOStream has no fileno.\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    188\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    189\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mwrite\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstring\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mUnsupportedOperation\u001b[0m: IOStream has no fileno."
     ]
    }
   ],
   "source": [
    "def spawn(prog, *args):                       # pass progname, cmdline args\n",
    "    stdinFd  = sys.stdin.fileno()             # get descriptors for streams\n",
    "    stdoutFd = sys.stdout.fileno()            # normally stdin=0, stdout=1\n",
    "\n",
    "    parentStdin, childStdout  = os.pipe()     # make two IPC pipe channels\n",
    "    childStdin,  parentStdout = os.pipe()     # pipe returns (inputfd, outoutfd)\n",
    "    pid = os.fork()                           # make a copy of this process\n",
    "    if pid:\n",
    "        os.close(childStdout)                 # in parent process after fork:\n",
    "        os.close(childStdin)                  # close child ends in parent\n",
    "        os.dup2(parentStdin,  stdinFd)        # my sys.stdin copy  = pipe1[0]\n",
    "        os.dup2(parentStdout, stdoutFd)       # my sys.stdout copy = pipe2[1]\n",
    "    else:\n",
    "        os.close(parentStdin)                 # in child process after fork:\n",
    "        os.close(parentStdout)                # close parent ends in child\n",
    "        os.dup2(childStdin,  stdinFd)         # my sys.stdin copy  = pipe2[0]\n",
    "        os.dup2(childStdout, stdoutFd)        # my sys.stdout copy = pipe1[1]\n",
    "        args = (prog,) + args\n",
    "        os.execvp(prog, args)                 # new program in this process\n",
    "        assert False, 'execvp failed!'        # os.exec call never returns here\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    mypid = os.getpid()\n",
    "    spawn('python', '-u', 'pipes-testchild.py', 'spam')     # fork child program\n",
    "\n",
    "    print('Hello 1 from parent', mypid)               # to child's stdin\n",
    "    sys.stdout.flush()                                # subvert stdio buffering\n",
    "    reply = input()                                   # from child's stdout\n",
    "    sys.stderr.write('Parent got: \"%s\"\\n' % reply)    # stderr not tied to pipe!\n",
    "\n",
    "    print('Hello 2 from parent', mypid)\n",
    "    sys.stdout.flush()\n",
    "    reply = sys.stdin.readline()\n",
    "    sys.stderr.write('Parent got: \"%s\"\\n' % reply[:-1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Child 3257 of 2729 got arg: \"-f\"\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ss\n",
      "Child 3257 got: [ss]\n",
      "rt\n",
      "Child 3257 got: [rt]\n"
     ]
    }
   ],
   "source": [
    "import os, time, sys\n",
    "mypid     = os.getpid()\n",
    "parentpid = os.getppid()\n",
    "sys.stderr.write('Child %d of %d got arg: \"%s\"\\n' % (mypid, parentpid, sys.argv[1]))\n",
    "for i in range(2):\n",
    "    time.sleep(3)              # make parent process wait by sleeping here\n",
    "    recv = raw_input()             # stdin tied to pipe: comes from parent's stdout\n",
    "    time.sleep(3)\n",
    "    send = 'Child %d got: [%s]' % (mypid, recv)\n",
    "    print send                # stdout tied to pipe: goes to parent's stdin\n",
    "    sys.stdout.flush()         # make sure it's sent now or else process blocks"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.1.5 Output stream buffering revisited: Deadlocks and flushes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Deadlock in general, though, is a bigger problem than we have space to address fully\n",
    "# here. On the other hand, if you know enough that you want to do IPC in Python, you’re\n",
    "# probably already a veteran of the deadlock wars."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2 Named Pipes (Fifos)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# named pipes; os.mkfifo is not available on Windows (without Cygwin); \n",
    "# there is no reason to fork here, since fifo file pipes are external \n",
    "# to processes--shared fds in parent/child processes are irrelevent;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import os, time, sys\n",
    "fifoname = '/tmp/pipefifo'                       # must open same name\n",
    "\n",
    "def child():\n",
    "    pipeout = os.open(fifoname, os.O_WRONLY)     # open fifo pipe file as fd\n",
    "    zzz = 0\n",
    "    while True:\n",
    "        time.sleep(zzz)\n",
    "        msg = ('Spam %03d\\n' % zzz).encode()     # binary as opened here\n",
    "        os.write(pipeout, msg)\n",
    "        zzz = (zzz+1) % 5\n",
    "\n",
    "def parent():\n",
    "    pipein = open(fifoname, 'r')                 # open fifo as text file object\n",
    "    while True:\n",
    "        line = pipein.readline()[:-1]            # blocks until data sent\n",
    "        print('Parent %d got \"%s\" at %s' % (os.getpid(), line, time.time()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "if __name__ == '__main__':\n",
    "    if not os.path.exists(fifoname):\n",
    "        os.mkfifo(fifoname)                      # create a named pipe file\n",
    "    if len(sys.argv) == 1:\n",
    "        parent()                                 # run as parent if no args\n",
    "    else:                                        # else run as child process\n",
    "        child()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3 Sockets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# sockets for cross-task communication: start threads to communicate over sockets;\n",
    "# independent programs can too, because sockets are system-wide, much like fifos;\n",
    "# see the GUI and Internet parts of the book for more realistic socket use cases;\n",
    "# some socket servers may also need to talk to clients in threads or processes;\n",
    "# sockets pass byte strings, but can be pickled objects or encoded Unicode text;\n",
    "# caveat: prints in threads may need to be synchronized if their output overlaps;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from socket import socket, AF_INET, SOCK_STREAM     # portable socket api\n",
    "\n",
    "port = 50008                 # port number identifies socket on machine\n",
    "host = 'localhost'           # server and client run on same local machine here\n",
    "\n",
    "def server():\n",
    "    sock = socket(AF_INET, SOCK_STREAM)         # ip addresses tcp connection\n",
    "    sock.bind(('', port))                       # bind to port on this machine\n",
    "    sock.listen(5)                              # allow up to 5 pending clients\n",
    "    while True:\n",
    "        conn, addr = sock.accept()              # wait for client to connect\n",
    "        data = conn.recv(1024)                  # read bytes data from this client\n",
    "        reply = 'server got: [%s]' % data       # conn is a new connected socket\n",
    "        conn.send(reply.encode())               # send bytes reply back to client\n",
    "\n",
    "def client(name):\n",
    "    sock = socket(AF_INET, SOCK_STREAM)\n",
    "    sock.connect((host, port))                  # connect to a socket port\n",
    "    sock.send(name.encode())                    # send bytes data to listener\n",
    "    reply = sock.recv(1024)                     # receive bytes data from listener\n",
    "    sock.close()                                # up to 1024 bytes in message\n",
    "    print('client got: [%s]' % reply)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "client got: [server got: [client4]]\n"
     ]
    }
   ],
   "source": [
    "if __name__ == '__main__':\n",
    "    from threading import Thread\n",
    "    sthread = Thread(target=server)\n",
    "    sthread.daemon = True                       # don't wait for server thread\n",
    "    sthread.start()                             # do wait for children to exit\n",
    "    for i in range(5): \n",
    "         Thread(target=client, args=('client%s' % i,)).start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4 Signals"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# # catch signals in Python; pass signal number N as a command-line arg,\n",
    "# # use a \"kill -N pid\" shell command to send this process a signal;  most\n",
    "# # signal handlers restored by Python after caught (see network scripting\n",
    "# # chapter for SIGCHLD details); on Windows, signal module is available,\n",
    "# # but it defines only a few signal types there, and os.kill is missing;\n",
    "\n",
    "import sys, signal, time\n",
    "def now(): return time.ctime(time.time())        # current time string\n",
    "\n",
    "def onSignal(signum, stackframe):                # python signal handler\n",
    "    print('Got signal', signum, 'at', now())     # most handlers stay in effect\n",
    "\n",
    "signum = int(sys.argv[1])\n",
    "signal.signal(signum, onSignal)                  # install signal handler\n",
    "while True: signal.pause()                       # wait for signals (or: pass)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('Setting at', 'Tue May 20 15:58:01 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:58:06 2014')\n",
      "('Setting at', 'Tue May 20 15:58:06 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:58:11 2014')\n",
      "('Setting at', 'Tue May 20 15:58:11 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:58:16 2014')\n",
      "('Setting at', 'Tue May 20 15:58:16 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:58:21 2014')\n",
      "('Setting at', 'Tue May 20 15:58:21 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:58:26 2014')\n",
      "('Setting at', 'Tue May 20 15:58:26 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:58:31 2014')\n",
      "('Setting at', 'Tue May 20 15:58:31 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:58:36 2014')\n",
      "('Setting at', 'Tue May 20 15:58:36 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:58:41 2014')\n",
      "('Setting at', 'Tue May 20 15:58:41 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:58:46 2014')\n",
      "('Setting at', 'Tue May 20 15:58:46 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:58:51 2014')\n",
      "('Setting at', 'Tue May 20 15:58:51 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:58:56 2014')\n",
      "('Setting at', 'Tue May 20 15:58:56 2014')\n",
      "('Got alarm', 14, 'at', 'Tue May 20 15:59:01 2014')\n",
      "('Setting at', 'Tue May 20 15:59:01 2014')\n"
     ]
    },
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m\n\u001b[1;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-32-120c29997daa>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m     13\u001b[0m     \u001b[0msignal\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msignal\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0msignal\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mSIGALRM\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0monSignal\u001b[0m\u001b[1;33m)\u001b[0m       \u001b[1;31m# install signal handler\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     14\u001b[0m     \u001b[0msignal\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0malarm\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;36m5\u001b[0m\u001b[1;33m)\u001b[0m                               \u001b[1;31m# do signal in 5 seconds\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 15\u001b[1;33m     \u001b[0msignal\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mpause\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m                                \u001b[1;31m# wait for signals\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "# set and catch alarm timeout signals in Python; time.sleep doesn't play\n",
    "# well with alarm (or signal in general in my Linux PC), so we call\n",
    "# signal.pause here to do nothing until a signal is received;\n",
    "\n",
    "import sys, signal, time\n",
    "def now(): return time.asctime()\n",
    "\n",
    "def onSignal(signum, stackframe):                 # python signal handler\n",
    "    print('Got alarm', signum, 'at', now())       # most handlers stay in effect\n",
    "\n",
    "while True:\n",
    "    print('Setting at', now())\n",
    "    signal.signal(signal.SIGALRM, onSignal)       # install signal handler\n",
    "    signal.alarm(5)                               # do signal in 5 seconds\n",
    "    signal.pause()                                # wait for signals"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# The multiprocessing Module"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1 The Basics: Processes and Locks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('Got alarm', 14, 'at', 'Tue May 20 15:59:06 2014')\n",
      "function call: name:__main__, pid:3257\n",
      "spawned child: name:__main__, pid:5196\n",
      "Main process exit.\n",
      "run process 0: name:__main__, pid:5204\n",
      "run process 1: name:__main__, pid:5205\n",
      "run process 2: name:__main__, pid:5206\n"
     ]
    }
   ],
   "source": [
    "\"\"\"\n",
    "multiprocess basics: Process works like threading.Thread, but \n",
    "runs function call in parallel in a process instead of a thread;\n",
    "locks can be used to synchronize, e.g. prints on some platforms;\n",
    "starts new interpreter on windows, forks a new process on unix;\n",
    "\"\"\"\n",
    "\n",
    "import os\n",
    "from multiprocessing import Process, Lock\n",
    "\n",
    "def whoami(label, lock):\n",
    "    msg = '%s: name:%s, pid:%s'\n",
    "    with lock:\n",
    "        print(msg % (label, __name__, os.getpid()))\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    lock = Lock()\n",
    "    whoami('function call', lock)\n",
    "\n",
    "    p = Process(target=whoami, args=('spawned child', lock))\n",
    "    p.start()\n",
    "    p.join()\n",
    "\n",
    "    for i in range(5):\n",
    "        Process(target=whoami, args=(('run process %s' % i), lock)).start()\n",
    "\n",
    "    with lock:\n",
    "        print('Main process exit.')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2 IPC Tools: Pipes, Shared Memory, and Queues"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.1 multiprocessing pipes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Use multiprocess anonymous pipes to communicate. Returns 2 connection\n",
    "# object representing ends of the pipe: objects are sent on one end and\n",
    "# received on the other, though pipes are bidirectional by default\n",
    "\n",
    "from multiprocessing import Process, Pipe\n",
    "\n",
    "def sender(pipe):\n",
    "    pipe.send(['spam'] + [42, 'eggs'])\n",
    "    pipe.close()\n",
    "\n",
    "def talker(pipe):\n",
    "    pipe.send(dict(name='bob', spam=42))\n",
    "    reply = pipe.recv()\n",
    "    print 'talk got:', reply\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    parentEnd, childEnd = Pipe()\n",
    "    Process(target=sender,args=(childEnd,)).start()\n",
    "    print 'parent got:', parentEnd.recv()\n",
    "    parentEnd.close()\n",
    "\n",
    "    parentEnd, childEnd = Pipe()\n",
    "    child = Process(target=talker,args=(parentEnd,))\n",
    "    child.start()\n",
    "    print 'parent got:', parentEnd.recv()\n",
    "    parentEnd.send([x *2 for x in 'spam'])\n",
    "    child.join()\n",
    "    print 'parent exit'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.2 Shared memory and globals"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "\"\"\"\n",
    "Use multiprocess shared memory objects to communicate.\n",
    "Passed objects are shared, but globals are not on Windows.\n",
    "Last test here reflects common use case: distributing work.\n",
    "\"\"\"\n",
    "\n",
    "import os\n",
    "from multiprocessing import Process, Value, Array\n",
    "\n",
    "procs = 3\n",
    "count = 0    # per-process globals, not shared \n",
    "\n",
    "def showdata(label, val, arr):\n",
    "    \"\"\"\n",
    "    print data values in this process\n",
    "    \"\"\"\n",
    "    msg = '%-12s: pid:%4s, global:%s, value:%s, array:%s'\n",
    "    print(msg % (label, os.getpid(), count, val.value, list(arr)))\n",
    "\n",
    "def updater(val, arr):\n",
    "    \"\"\"\n",
    "    communicate via shared memory\n",
    "    \"\"\"\n",
    "    global count\n",
    "    count += 1                         # global count not shared\n",
    "    val.value += 1                     # passed in objects are\n",
    "    for i in range(3): arr[i] += 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "parent start: pid:3257, global:0, value:0, array:[0.0, 0.0, 0.0]\n",
      "\n",
      "loop1 (updates in parent, serial children)...\n",
      "child       : pid:5361, global:0, value:0, array:[0.0, 0.0, 0.0]\n",
      "process 0   : pid:5369, global:1, value:1, array:[1.0, 0.0, 0.0]\n",
      "process 1   : pid:5377, global:2, value:2, array:[1.0, 1.0, 0.0]\n",
      "\n",
      "loop2 (updates in parent, parallel children)...\n",
      "process 2   : pid:5385, global:3, value:3, array:[1.0, 1.0, 1.0]\n",
      "\n",
      "loop3 (updates in serial children)...\n",
      "process 0   : pid:5393, global:4, value:6, array:[2.0, 2.0, 2.0]\n",
      "process 2   : pid:5397, global:6, value:6, array:[2.0, 2.0, 2.0]\n",
      "process 1   : pid:5394, global:5, value:6, array:[2.0, 2.0, 2.0]\n",
      "parent temp : pid:3257, global:6, value:9, array:[5.0, 5.0, 5.0]\n",
      "\n",
      "loop4 (updates in parallel children)...\n",
      "parent end  : pid:3257, global:6, value:12, array:[8.0, 8.0, 8.0]\n"
     ]
    }
   ],
   "source": [
    "if __name__ == '__main__':\n",
    "    scalar = Value('i', 0)             # shared memory: process/thread safe\n",
    "    vector = Array('d', procs)         # type codes from ctypes: int, double\n",
    "\n",
    "    # show start value in parent process\n",
    "    showdata('parent start', scalar, vector)\n",
    "\n",
    "    # spawn child, pass in shared memory\n",
    "    p = Process(target=showdata, args=('child ', scalar, vector))\n",
    "    p.start(); p.join()\n",
    "\n",
    "    # pass in shared memory updated in parent, wait for each to finish\n",
    "    # each child sees updates in parent so far for args (but not global)\n",
    "\n",
    "    print('\\nloop1 (updates in parent, serial children)...')\n",
    "    for i in range(procs):\n",
    "        count += 1\n",
    "        scalar.value += 1\n",
    "        vector[i] += 1\n",
    "        p = Process(target=showdata, args=(('process %s' % i), scalar, vector))\n",
    "        p.start(); p.join()\n",
    "\n",
    "    # same as prior, but allow children to run in parallel\n",
    "    # all see the last iteration's result because all share objects\n",
    "\n",
    "    print('\\nloop2 (updates in parent, parallel children)...')\n",
    "    ps = []\n",
    "    for i in range(procs):\n",
    "        count += 1\n",
    "        scalar.value += 1\n",
    "        vector[i] += 1\n",
    "        p = Process(target=showdata, args=(('process %s' % i), scalar, vector))\n",
    "        p.start()\n",
    "        ps.append(p)\n",
    "    for p in ps: p.join()\n",
    "\n",
    "    # shared memory updated in spawned children, wait for each \n",
    "\n",
    "    print('\\nloop3 (updates in serial children)...')\n",
    "    for i in range(procs):\n",
    "        p = Process(target=updater, args=(scalar, vector))\n",
    "        p.start()\n",
    "        p.join()\n",
    "    showdata('parent temp', scalar, vector)\n",
    "    \n",
    "    # same, but allow children to update in parallel\n",
    "\n",
    "    ps = []\n",
    "    print('\\nloop4 (updates in parallel children)...')\n",
    "    for i in range(procs):\n",
    "        p = Process(target=updater, args=(scalar, vector))\n",
    "        p.start()\n",
    "        ps.append(p)\n",
    "    for p in ps: p.join()\n",
    "    # global count=6 in parent only  \n",
    "    # show final results here\n",
    "    # scalar=12:  +6 parent, +6 in 6 children\n",
    "    showdata('parent end', scalar, vector) # array[i]=8: +2 parent, +6 in 6 children"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.3 Queues and subclassing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "start 3257\n",
      "no data...\n",
      "no data...\n",
      "no data...\n",
      "no data...\n",
      "no data...\n",
      "no data... @ 5847 1\n",
      "\n",
      "posted: @ 5848 101\n",
      " @ 5849 1001\n",
      " 5848\n",
      "posted: 5847\n",
      "posted: 5849\n",
      "no data...\n",
      "no data...\n",
      "posted: @ @ 5847 2\n",
      " 5848 102\n",
      " 5847\n",
      "posted: @ 5849 1002\n",
      " 5848\n",
      "posted: 5849\n",
      "no data...\n",
      "no data...\n",
      "no data...\n",
      "posted: @ @ @ 5847 3\n",
      " 5848 103\n",
      " 5849 1003\n",
      " @ 5847 -\n",
      " @ 5848 -\n",
      " @ 5849 -\n",
      " 5847\n",
      "posted: 5848\n",
      "posted: 5849\n",
      "finish 3257 0\n"
     ]
    }
   ],
   "source": [
    "import os, time, Queue\n",
    "from multiprocessing import Process, Queue as queue\n",
    "class Counter(Process):\n",
    "    \"\"\"docstring for Counter\"\"\"\n",
    "    label = ' @'\n",
    "    def __init__(self, start, Queue):\n",
    "        self.state = start\n",
    "        self.post  = Queue\n",
    "        Process.__init__(self)\n",
    "    def run(self):\n",
    "        for i in range(3):\n",
    "            time.sleep(3)\n",
    "            self.state += 1\n",
    "            print self.label, self.pid, self.state\n",
    "            self.post.put(self.pid, self.state)\n",
    "        print self.label, self.pid, '-'\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    print 'start', os.getpid()\n",
    "    expected = 9\n",
    "    post = queue()\n",
    "    p = Counter(0,post)\n",
    "    q = Counter(100,post)\n",
    "    r = Counter(1000,post)\n",
    "    p.start(); q.start(); r.start()\n",
    "\n",
    "    while expected:                             # parent consumes data on queue\n",
    "        time.sleep(0.5)                         # this is essentially like a GUI,\n",
    "        try:                                    # though GUIs often use threads\n",
    "            data = post.get(block=False)\n",
    "        except Queue.Empty:\n",
    "            print 'no data...'\n",
    "        else:\n",
    "            print 'posted:', data\n",
    "            expected -= 1\n",
    "\n",
    "    p.join(); q.join(); r.join()                # must get before join putter\n",
    "    print 'finish', os.getpid(), r.exitcode   # exitcode is child exit status"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3 Starting Independent Programs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "parent exit\n"
     ]
    }
   ],
   "source": [
    "\"Use multiprocessing to start independent programs, os.fork or not\"\n",
    "import os\n",
    "from multiprocessing import Process\n",
    "\n",
    "def runp(arg):\n",
    "    os.execlp('python', 'python', 'child.py', str(arg))\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    for i in range(5):\n",
    "        Process(target=runp, args=(i,)).start()\n",
    "    print('parent exit')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4 process pools"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4]\n",
      "[4, 4]\n",
      "[1, 2, 4, 8, 16, 32, 64, 128, 256, 512]\n",
      "[256, 512]\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Pool\n",
    "\n",
    "def powers(x):\n",
    "    return 2 ** x\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    workers = Pool(processes=5)\n",
    "\n",
    "    results = workers.map(powers,[2]*100)\n",
    "    print results[:16]\n",
    "    print results[-2:]\n",
    "\n",
    "    results = workers.map(powers,range(10))\n",
    "    print results[:16]\n",
    "    print results[-2:]"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}
